package beautiful.butterfly.drds.data_exchange.taskgroup;


import beautiful.butterfly.drds.data_exchange.constant.Constants;
import beautiful.butterfly.drds.data_exchange.constant.State;
import beautiful.butterfly.drds.data_exchange.constant.TaskType;
import beautiful.butterfly.drds.data_exchange.data_exchange.exchanger.Produce;
import beautiful.butterfly.drds.data_exchange.data_exchange.exchanger.QueueExchanger;
import beautiful.butterfly.drds.data_exchange.data_exchange.record_queue.MemoryRecordQueue;
import beautiful.butterfly.drds.data_exchange.data_exchange.record_queue.RecordQueue;
import beautiful.butterfly.drds.data_exchange.error_code.DataExchangeException;
import beautiful.butterfly.drds.data_exchange.error_code.FrameworkErrorCode;
import beautiful.butterfly.drds.data_exchange.plugin_collector.AbstractTaskPluginCollector;
import beautiful.butterfly.drds.data_exchange.plugin_collector.TaskPluginCollector;
import beautiful.butterfly.drds.data_exchange.report.Message;
import beautiful.butterfly.drds.data_exchange.taskgroup.runner.ReadeTask;
import beautiful.butterfly.drds.data_exchange.taskgroup.runner.Task;
import beautiful.butterfly.drds.data_exchange.taskgroup.runner.WriteTask;
import beautiful.butterfly.drds.data_exchange.util.Configuration;
import org.apache.commons.lang3.Validate;

/**
 * TaskExecutor是一个完整task的执行器
 * 其中包括1：1的reader和writer
 */
class TaskExecutor
{
    private TaskGroupContainer taskGroupContainer;
    private Configuration taskConfig;

    private int taskId;

    private int attemptCount;

    private RecordQueue recordQueue;

    private Thread readerThread;

    private Thread writerThread;

    private ReadeTask readeTask;

    private WriteTask writeTask;

    /**
     * 该处的taskCommunication在多处用到：
     * 1. recordQueue
     * 2. readerRunner和writerRunner
     * 3. reader和writer的taskPluginCollector
     */
    private Message message;

    public TaskExecutor(TaskGroupContainer taskGroupContainer, Configuration taskConf, int attemptCount)
    {
        this.taskGroupContainer = taskGroupContainer;
        // 获取该taskExecutor的配置
        this.taskConfig = taskConf;
        Validate.isTrue(null != this.taskConfig.getConfigureDataMap(Constants.job_reader)
                        && null != this.taskConfig.getConfigureDataMap(Constants.job_writer),
                "[reade|write]的插件参数不能为空!");

        // 得到taskId
        this.taskId = this.taskConfig.getInt(Constants.task_id);
        this.attemptCount = attemptCount;

        /**
         * 由taskId得到该taskExecutor的Communication
         * 要传给readerRunner和writerRunner，同时要传给channel作统计用
         */
        this.message = taskGroupContainer.containerCommunicator
                .getCommunication(taskId);
        Validate.notNull(this.message,
                String.format("taskId[%d]的Communication没有注册过", taskId));


        this.recordQueue = new MemoryRecordQueue(taskGroupContainer.configuration);

        this.recordQueue.setMessage(this.message);


        /**
         * 生成writerThread
         */
        writeTask = (WriteTask) newTask(TaskType.write);
        this.writerThread = new Thread(writeTask, String.format("%d-%d-%d-write", taskGroupContainer.jobId, taskGroupContainer.taskGroupId, this.taskId));


        /**
         * 生成readerThread
         */
        readeTask = (ReadeTask) newTask(TaskType.reade);
        this.readerThread = new Thread(readeTask, String.format("%d-%d-%d-reade", taskGroupContainer.jobId, taskGroupContainer.taskGroupId, this.taskId));

    }

    public void doStart()
    {
        this.writerThread.start();

        // reader没有起来，writer不可能结束
        if (!this.writerThread.isAlive() || this.message.getState() == State.failed)
        {
            throw DataExchangeException.asDataExchangeException(
                    FrameworkErrorCode.RUNTIME_ERROR,
                    this.message.getThrowable());
        }

        this.readerThread.start();

        // 这里reader可能很快结束
        if (!this.readerThread.isAlive() && this.message.getState() == State.failed)
        {
            // 这里有可能出现Reader线上启动即挂情况 对于这类情况 需要立刻抛出异常
            throw DataExchangeException.asDataExchangeException(
                    FrameworkErrorCode.RUNTIME_ERROR,
                    this.message.getThrowable());
        }

    }


    public Task newTask(TaskType taskType)
    {
        Task task = null;
        AbstractTaskPluginCollector pluginCollector;

        switch (taskType)
        {
            case reade:
                task = new ReadeTask(null);//
                task.setConfiguration(this.taskConfig.getConfigureDataMap(
                        Constants.job_reader_parameter));


                pluginCollector = new TaskPluginCollector(taskGroupContainer.configuration, this.message,
                        TaskType.reade);


                Produce produce = new QueueExchanger(this.recordQueue, pluginCollector);

                ((ReadeTask) task).setProduce(produce);

                /**
                 * 设置taskPlugin的collector，用来处理脏数据和job/task通信
                 */
                task.setTaskPluginCollector(pluginCollector);
                break;
            case write:

                task = new WriteTask(null);//
                task.setConfiguration(this.taskConfig
                        .getConfigureDataMap(Constants.job_writer_parameter));

                pluginCollector = new TaskPluginCollector(taskGroupContainer.configuration, this.message,
                        TaskType.write);
                ((WriteTask) task).setConsume(new QueueExchanger(
                        this.recordQueue, pluginCollector));
                /**
                 * 设置taskPlugin的collector，用来处理脏数据和job/task通信
                 */
                task.setTaskPluginCollector(pluginCollector);
                break;
            default:
                throw DataExchangeException.asDataExchangeException(FrameworkErrorCode.ARGUMENT_ERROR, "Cant newTask for:" + taskType);
        }

        task.setTaskGroupId(taskGroupContainer.taskGroupId);
        task.setTaskId(this.taskId);
        task.setMessage(this.message);

        return task;
    }

    // 检查任务是否结束
    public boolean isTaskFinished()
    {
        // 如果reader 或 writer没有完成工作，那么直接返回工作没有完成
        if (readerThread.isAlive() || writerThread.isAlive())
        {
            return false;
        }

        if (message == null || !message.isFinished())
        {
            return false;
        }

        return true;
    }

    public int getTaskId()
    {
        return taskId;
    }

    public long getTimeStamp()
    {
        return message.getTimestamp();
    }

    public int getAttemptCount()
    {
        return attemptCount;
    }

    public boolean supportFailOver()
    {
        return writeTask.supportFailOver();
    }

    public void shutdown()
    {
        writeTask.shutdown();
        readeTask.shutdown();
        if (writerThread.isAlive())
        {
            writerThread.interrupt();
        }
        if (readerThread.isAlive())
        {
            readerThread.interrupt();
        }
    }

    public boolean isShutdown()
    {
        return !readerThread.isAlive() && !writerThread.isAlive();
    }
}
